import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.hash.MurmurHash;


public class Step0 {


	/** Input: (<obj>, "url|url_1|..|url_n")
	 *  where obj: 	     not significant
	 * 		  url: 	     url of a node, let's call it j
	 * 		  url_1..n:  urls of neighbours of j
	 * 	
	 *  Output: (<id>, "url|zeta|k|node_1|..|node_n") 
	 *  where id:		 hash of j's url
	 *  	  url:		 url of j
	 *  	  zeta:		 initialized to 0, is the number of visits to j
	 *  	  k:		 numerber of live coupons, must be c*log(n) 
	 *  	  node_1..n: hashes of j's neighbours
	 */
	//During this step, the graph file is read and coupons are created
	public static class step0_mapper extends Mapper<Object, Text, Text, Text>{
		private Text outputkey = new Text();
		private Text outputvalue = new Text();
		
		public void map(Object key, Text value, Context context) 
				throws IOException, InterruptedException{

			if (value.toString().equals("")) return;
			//we use '|' such as delimiter
			String[] tokens = value.toString().split("[|]");
			StringBuilder valuebuilder = new StringBuilder();
			//murmur hash is a distributed, fast hash function of hadoop
			MurmurHash h = new MurmurHash();
			
			long k = context.getConfiguration().getLong("clogn", 0);
			//a local counter is incremented counting the total number of coupons
			context.getCounter(PR_COUNTERS.BORN).increment(k);

			//The key is the hash of the first url (j's url).
			outputkey.set(new Integer(h.hash(tokens[0].getBytes())).toString());
			//if (outputkey.toString().equals("-1285986640"))
			//	System.out.println("Eccolo: " + value.toString());			

			//Initialize the value to "url|0|k|"
			valuebuilder.append(tokens[0] + "|0|" + k + "|");
			
			//Append to the value all the hashes of j's neighbours
			for(int i=1; i<tokens.length; i++){
				valuebuilder.append(h.hash(tokens[i].getBytes()) + "|");	
			}
			
			outputvalue.set(valuebuilder.toString());
			context.write(outputkey, outputvalue);
		}
	}
	
	
	
	
	public static void main(String [] args) throws Exception{
		
		Job job0, job1, job2;
		long n, c, clogn, step1_it=0, jobtime, nclogn, totaltime;
		long deads=0;
		float epsilon;
		int job1_reducers;
		Configuration conf;
		String inputpath, outputpath, tmp, oprefix;
		FileSystem fs = FileSystem.get(new Configuration());
		
		//Input parameters
		inputpath = args[0];
		oprefix = args[1];
		c = Long.parseLong(args[2]);
		n = Long.parseLong(args[3]);
		job1_reducers = Integer.parseInt(args[5]);
		epsilon = Float.parseFloat(args[4]);
		clogn = (long) Math.round(c*Math.log(n));		

		System.out.println("c: " + c + "; n: " + n + " clogn: " + clogn + "; nclogn: " + (n*clogn)); 
	
		//-------Job0------
		outputpath = oprefix + "_step_0";
		job0 = new Job(new Configuration(), "pr_step0");
		job0.setJarByClass(Step0.class);
		
		job0.setOutputKeyClass(Text.class);
		job0.setOutputValueClass(Text.class);
		
		job0.setMapperClass(step0_mapper.class);
                
		job0.setNumReduceTasks(0);	
	
		FileInputFormat.addInputPath(job0, new Path(inputpath));
		FileOutputFormat.setOutputPath(job0, new Path(outputpath));
				
		job0.getConfiguration().setLong("clogn", clogn);

		jobtime = System.currentTimeMillis();
		totaltime = jobtime;
		job0.waitForCompletion(true);
		jobtime = System.currentTimeMillis() - jobtime;
		
		System.out.println("Job1: " + jobtime + "ms");
		
		//-------Job1------
		conf = new Configuration();
		conf.setFloat("epsilon", epsilon);
		inputpath = outputpath;
		outputpath = oprefix + "step_1";
		nclogn=job0.getCounters().findCounter(PR_COUNTERS.BORN).getValue();
		
		/*For each node of the graph clogn coupons are created. In total we 
		have nclogn coupons: the loop must termite when all coupond are died. 
		The variable deads sum the deads at each job execution. */
		//System.out.println("Deads: " + deads + "; n:" + n + "; clogn: " + clogn);
		//System.out.println("nclogn: " + (n*clogn));		

		//System.out.println("nclogn: " + (n*clogn) + "; n: " + n + "; clogn: " + clogn);		

		long cprev = nclogn;
		while (deads<nclogn){
			//Create and configure the job
			job1 = new Job(conf, "pr_step1_" + step1_it);
			job1.setJarByClass(Step0.class);
			if (job1_reducers>0) job1.setNumReduceTasks(job1_reducers);		
			job1.setOutputKeyClass(Text.class);
			job1.setOutputValueClass(Text.class);
			
			job1.setMapperClass(Step1.step1_mapper.class);
			//job1.setCombinerClass(Step1.step1_combiner.class);
			job1.setReducerClass(Step1.step1_reducer.class);
			
			FileInputFormat.addInputPath(job1, new Path(inputpath));
			FileOutputFormat.setOutputPath(job1, new Path(outputpath));
		
			//execute the job
			jobtime = System.currentTimeMillis();
			job1.waitForCompletion(true);
			jobtime = System.currentTimeMillis() - jobtime;
			System.out.println("Job1, it: " + step1_it + ": " + jobtime + "ms");
			//delete the inputpath
			fs.delete(new Path(inputpath), true);
			
			//swap inputpath and outputpath
			tmp = inputpath;
			inputpath = outputpath;
			outputpath = tmp;
			
			//Increase the dead counter
			long cdeads = job1.getCounters().findCounter(PR_COUNTERS.DEADS).getValue();
			deads += cdeads;
			long currentc = job1.getCounters().findCounter(PR_COUNTERS.CURRENT).getValue();
			long error = cprev-cdeads-currentc;
			step1_it++;
			System.out.println("deads: " + deads + "; cdeads: " + cdeads + "; nclogn: " + nclogn + "; survived: " + currentc + "; error: " + error);
			if (error!=0) System.out.println("ERROR: lost coupons: " + error);
			cprev = currentc;
		}
		
		if (deads>n*clogn) System.out.println("WARNING: deads>coupons");

		//-------Job2------
		outputpath = oprefix + "_final";
		job2 = new Job(new Configuration(), "pr_step2");
		job2.setJarByClass(Step2.class);
		
		job2.setOutputKeyClass(Text.class);
		job2.setOutputValueClass(Text.class);
		
		job2.setMapperClass(Step2.step2_mapper.class);
		
		FileInputFormat.addInputPath(job2, new Path(inputpath));
		FileOutputFormat.setOutputPath(job2, new Path(outputpath));

		job2.getConfiguration().setLong("cnlogn", nclogn);
		job2.getConfiguration().setFloat("epsilon", epsilon);

		job2.setNumReduceTasks(0);		

		jobtime = System.currentTimeMillis();
		job2.waitForCompletion(true);
		jobtime = System.currentTimeMillis() - jobtime;
		totaltime = System.currentTimeMillis() - totaltime;
		System.out.println("Job2: " + jobtime + "ms");
		System.out.println("Total: " + totaltime + "ms");
	}
}
